package com.atguigu.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.OperatorStateStore;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 *
 * @date 2024/11/20 11:13
 *
 * 算子状态 - 联合列表状态
 *
 * 列表状态和 联合列表状态的区别:
 *    从使用上来说。两者没有太大区别， 本质上都是列表状态。
 *    当程序重启，还原状态的时候， 列表状态还原后的状态与程序失败前保持一致。  联合列表状态还原后的状态是 将失败前每个并行实例的状态联合到一起给到每个并行实例。
 *
 *
 * 联合列表状态的应用场景:   KafkaSource
 *
 *    失败前:
 *            Source       并行实例     消费的分区    offset
 *         KafkaSource  ->   P0    ->    P0    ->   100
 *         KafkaSource  ->   P1    ->    P1    ->   101
 *         KafkaSource  ->   P2    ->    P2    ->   99
 *         KafkaSource  ->   P3    ->    P3    ->   102
 *
 *    失败重启后:
 *            Source       并行实例     消费的分区     offset
 *          KafkaSource  ->   P0    ->    P1    ->   [P0 -> 100 , P1 -> 101  , P2 -> 99 , P3 -> 102 ]
            KafkaSource  ->   P1    ->    P3    ->   [P0 -> 100 , P1 -> 101  , P2 -> 99 , P3 -> 102 ]
            KafkaSource  ->   P2    ->    P0    ->   [P0 -> 100 , P1 -> 101  , P2 -> 99 , P3 -> 102 ]
            KafkaSource  ->   P3    ->    P2    ->   [P0 -> 100 , P1 -> 101  , P2 -> 99 , P3 -> 102 ]
 *
 *
 */
//todo 注：先写未测试代码

public class Flink03_OperatorUnionListState {
     public static void main(String[] args) {
         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
         env.setParallelism(1);

         //开启检查点，程序失败后会无限重启
         env.enableCheckpointing(2000L);
         //对接数据源
         DataStreamSource<String> ds = env.socketTextStream("hadoop102", 9999);

         //为啥通过map算子记录输入的每个数据,测试算子的状态
         ds.map(
                 new OperatorUnionListStateMapFunction()//转化数据
         ).setParallelism(2)
         .addSink(
                 new MysinkFunction() //将数据写出去，复习写出算子！！
         );



         try {
                 env.execute();
             } catch (Exception e) {
                 throw new RuntimeException(e);
             }
         }

    private static class OperatorUnionListStateMapFunction implements MapFunction<String,String>, CheckpointedFunction {

         //首先声明状态值
         //定义一个集合，记录map中的所有输入数据
        //原始状态,这里为啥注掉？因为原始状态重启后状态值会清空
        //List<String>dataList = new ArrayList<String>();

        //联合列表状态
        private ListState<String> listState;

        @Override
        public String map(String value) throws Exception {
            //往状态中添加数据
            listState.add(value);
            //从状态中获取数据
            return listState.get().toString(); //从listState中获取当前状态，并将其转换为字符串类型后返回。
        }

        @Override
        public void snapshotState(FunctionSnapshotContext functionSnapshotContext) throws Exception {
            //表示正在拍摄状态快照
            System.out.println("snapshotState........");

        }

        /*
        * 初始化状态，状态的获取 以及 程序重启时状态的还原
        *
        * 程序启动时会调用一次
        *
        * */

        @Override
        public void initializeState(FunctionInitializationContext context) throws Exception {
            System.out.println("initializeState........");

            //初始化状态
            OperatorStateStore operatorStateStore = context.getOperatorStateStore();
            ListStateDescriptor<String> listStateDesc = new ListStateDescriptor<>("listStateDesc", Types.STRING);

            listState = operatorStateStore.getUnionListState(listStateDesc);


        }
    }

    private static class MysinkFunction implements SinkFunction<String> {
        @Override
        public void invoke(String value, Context context) throws Exception {

            if (value.contains("x")){
                //模拟异常
                throw new RuntimeException("抛出了异常");
            }

            System.out.println("MYPRINT>" + value);
        }
    }
}
